package com.wyz.socket.core.impl.dispatcher;

import com.wyz.socket.core.api.dispatcher.ReceiveDispatcher;
import com.wyz.socket.core.base.IoArgs;
import com.wyz.socket.core.api.Receiver;
import com.wyz.socket.core.base.packet.AbstractReceivePacket;

import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncReceiveDispatcher implements ReceiveDispatcher, IoArgs.IoBufferEventProcessor, AsyncPacketWriter.PacketProvider {
    private final AtomicBoolean isClosed = new AtomicBoolean();
    private final Receiver receiver;
    private final ReceiveCallback callback;
    private final AsyncPacketWriter writer;
    private volatile long activeTime = System.currentTimeMillis();

    public AsyncReceiveDispatcher(Receiver receiver, ReceiveCallback callback) {
        this.receiver = receiver;
        this.receiver.registerSocketListener(this);
        this.callback = callback;
        this.writer = new AsyncPacketWriter(this);
    }

    @Override
    public void start() {
        this.receiver.setReceiveListener(this);
        registerReceive();
    }

    @Override
    public void stop() {
        this.receiver.setReceiveListener(null);
    }

    private void registerReceive() {
        try {
            this.receiver.postReceiveAsync();
        } catch (Exception e) {
            e.printStackTrace();
            close();
        }
    }

    @Override
    public long getLastActiveTime() {
        return activeTime;
    }

    @Override
    public void onstart() {
        activeTime = System.currentTimeMillis();
    }

    @Override
    public IoArgs providerIoArgs() {
        IoArgs ioArgs = writer.providerIoArgs();
        ioArgs.startWriting();
        return ioArgs;
    }

    @Override
    public boolean onConsumeFailed(Throwable e) {
        if (e instanceof IOException) {
            return true;
        }
        return false;
    }

    @Override
    public boolean onConsumeCompleted(IoArgs args) {
        final AtomicBoolean isClosed = this.isClosed;
        final AsyncPacketWriter writer = this.writer;
        args.endWriting();
        do {
            writer.consumeIoArgs(args);
        } while (args.hasRemaining() && !isClosed.get());

        return !isClosed.get();
    }

    @Override
    public void onEnd() {

    }

    @Override
    public AbstractReceivePacket<?, ?> providerPacket(int length, byte type, byte[] headerInfo) {
        return callback.onArriyedNewPacket(type, length, headerInfo);
    }

    @Override
    public void completeSendPacket(AbstractReceivePacket<?, ?> packet, boolean isSucceed) {
        try {
            packet.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        callback.onReceivePacketComplated(packet, isSucceed);
    }

    @Override
    public void onReceiveHeartbeat() {
        callback.onReceivedHeartbeat();
    }


    @Override
    public void close() {
        try {
            receiver.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public <T> void onClose(T t) {
        if(t instanceof SocketChannel){
            if(!((SocketChannel) t).isOpen()){
                if (isClosed.compareAndSet(false, true)) {
                    writer.close();
                }
            }
        }
    }
}
